package com.oops.config;


import com.oops.RabbitMQConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.retry.support.RetryTemplateBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Objects;
import java.util.UUID;

@Configuration
@Slf4j
public class RabbitConfig {

    @Bean
    public ConnectionFactory connectionFactory(RabbitProperties rabbitProperties){
        final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setHost(rabbitProperties.getHost());
        cachingConnectionFactory.setPort(rabbitProperties.getPort());
        cachingConnectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
        cachingConnectionFactory.setUsername(rabbitProperties.getUsername());
        cachingConnectionFactory.setPassword(rabbitProperties.getPassword());
        final ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(12);
        threadPoolTaskExecutor.setMaxPoolSize(12);
        threadPoolTaskExecutor.setThreadNamePrefix("rabbitmq-executor-");
        threadPoolTaskExecutor.initialize();
        cachingConnectionFactory.setExecutor(threadPoolTaskExecutor);
        cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);

        return cachingConnectionFactory;
    }

    @Bean(value = "orderBizQueue")
    public Queue orderBizQueue() {
        // rmq可以从两个维度设置过期时间，队列、消息。此处从队列进行设置。
        final HashMap<String, Object> objectObjectHashMap = new HashMap<>();
        // 设置队列消息的ttl
        objectObjectHashMap.put("x-message-ttl", 20 * 1000);
        //设置队列的空闲存活时间（如果没有消费者，这个队列可以存活的时间）
        // objectObjectHashMap.put("x-expires", 60 * 1000);
        //设置队列的死信交换器（当消息到达ttl后会被路由到这个交换器）
        objectObjectHashMap.put("x-dead-letter-exchange",RabbitMQConstants.ORDER_DLX_EXCHANGE_NAME);
        //死信交换器的路由key（如果不指定则使用原来的routing-key）
        objectObjectHashMap.put("x-dead-letter-routing-key",RabbitMQConstants.ORDER_DLX_ROUTING_KEY);

        return new Queue(RabbitMQConstants.ORDER_BIZ_QUEUE_NAME, true, false, false, objectObjectHashMap);
    }

    @Bean(value = "orderDlxQueue")
    public Queue orderDlxQueue() {
        return new Queue(RabbitMQConstants.ORDER_DLX_QUEUE_NAME, true, false, false, null);
    }

    @Bean(value = "orderBizExchange")
    public Exchange exchange() {
        return new TopicExchange(RabbitMQConstants.ORDER_BIZ_EXCHANGE_NAME, true, false, null);
    }
    @Bean(value = "orderDlxExchange")
    public Exchange orderDlxExchange() {
        return new DirectExchange(RabbitMQConstants.ORDER_DLX_EXCHANGE_NAME, true, false, null);
    }

    @Bean(value = "orderBizBinding")
    public Binding orderBizBinding() {
        return new Binding(RabbitMQConstants.ORDER_BIZ_QUEUE_NAME, Binding.DestinationType.QUEUE, RabbitMQConstants.ORDER_BIZ_EXCHANGE_NAME, RabbitMQConstants.ORDER_BIZ_ROUTING_KEY, null);
    }

    @Bean(value = "orderDlxBinding")
    public Binding orderDlxBinding() {
        return new Binding(RabbitMQConstants.ORDER_DLX_QUEUE_NAME, Binding.DestinationType.QUEUE, RabbitMQConstants.ORDER_DLX_EXCHANGE_NAME, RabbitMQConstants.ORDER_DLX_ROUTING_KEY, null);
    }

    @Bean
    MessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter){
        final RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //消息发送失败，重新回到队列中。yml需要配置publisher-returns
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setMessageConverter(messageConverter);



        // 如果消息没有到exchange,则confirm回调,ack=false
        // 如果消息到达exchange,则confirm回调,ack=true
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            log.info("Thread.currentThread().getName() -> {}",Thread.currentThread().getName());
            assert correlationData != null;
            if (ack) {
                log.info("发送消息到交换器成功，ID为:{}  返回消息为:{}", correlationData.getId(), Objects.requireNonNull(correlationData.getReturned()));
            } else {
                log.info("发送消息到交换器失败，ID为:{}，原因是:{}", correlationData.getId(), cause);

            }
        });
        // exchange到queue成功,则不回调return
        // exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
        rabbitTemplate.setReturnsCallback(returned -> log.info("消息路由到队列的结果 returned : {}", returned));

        return rabbitTemplate;
    }
}